package com.ly.kafka.config;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ContainerProperties;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String springKafkaBootstrapServers;


    @Autowired
    private ConsumerFactory consumerFactory;

    /**
     * @describe：注入kafkaAdminClient，使用springboot注入报错
     * @author：李勇
     * @date：2023/4/11
     * @param
     * @return：org.apache.kafka.clients.admin.AdminClient
     */
    @Bean
    public AdminClient kafkaAdminClient() {
        Map<String, Object> props = new HashMap<>(1);
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, springKafkaBootstrapServers);
        return KafkaAdminClient.create(props);
    }
    
    /**
     * @describe：配置kafka自定义监听工厂batch类型
     * @author：李勇
     * @date：2023/4/14 
     * @param
     * @return：org.springframework.kafka.config.KafkaListenerContainerFactory<?>
     */       
    @Bean
    public KafkaListenerContainerFactory<?> batchFactory(){
        ConcurrentKafkaListenerContainerFactory<Integer,String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(5);
        //1秒拉取一次 poll
        factory.getContainerProperties().setPollTimeout(1000);
        factory.setBatchListener(true);//设置为批量消费，每个批次数量在Kafka配置参数中设置
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//设置手动提交ackMode
        return factory;
    }



    // 新建一个异常处理器，用@Bean注入
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return (message, exception, consumer) -> {
            //这里可以做一些业务
            System.out.println("消费异常："+message.getPayload());
            System.out.println("异常信息："+exception.getMessage());
            return null;
        };
    }


    /**
     * @describe：消息过滤器可以是多个,例如filterContainerFactory,filterContainerFactory1,filterContainerFactory2这种.
     * 消息过滤器可以在消息抵达consumer之前被拦截，在实际应用中，我们可以根据自己的业务逻辑，
     * 筛选出需要的信息再交由KafkaListener处理，不需要的消息则过滤掉。
     * @author：李勇
     * @date：2023/4/13
     * @param
     * @return：org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        // 被过滤的消息将被丢弃
        factory.setAckDiscarded(true);

        //如果监听者是批量消费的 则必须设置下面两个参数 如果不是 则不需要设置 因为配置文件读取不到
        factory.setBatchListener(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//设置手动提交ackMode

        // 消息过滤策略
        factory.setRecordFilterStrategy(consumerRecord -> {
            //处理拦截业务。。。。。
            if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {
                return false;
            }
            //返回true消息则被过滤
            return true;
        });
        return factory;
    }


    /**
     * @describe：监听器容器工厂(设置禁止KafkaListener自启动),也可以是多个delayContainerFactory,delayContainerFactory1...
     *            配合定时任务等业务启动消费者监听
     * @author：李勇
     * @date：2023/4/14
     * @param
     * @return：org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
        container.setConsumerFactory(consumerFactory);

        //如果监听者是批量消费的 则必须设置下面两个参数 如果不是 则不需要设置 因为配置文件读取不到
        container.setBatchListener(true);
        container.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//设置手动提交ackMode

        //禁止KafkaListener自启动
        container.setAutoStartup(false);
        return container;
    }











}
